package top.wuchu.rpc.serializer;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.wuchu.rpc.entity.RpcRequest;
import top.wuchu.rpc.entity.RpcResponse;
import top.wuchu.rpc.enumeration.SerializerCode;
import top.wuchu.rpc.exception.SerializeException;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * Kryo序列化器
 *
 * @author Wuchu
 */
public class KryoSerializer implements CommonSerializer {

    private static final Logger logger = LoggerFactory.getLogger(KryoSerializer.class);

    /**
     * 通过调用 `ThreadLocal` 的 `withInitial()` 方法，传入一个 `Lambda` 表达式，初始化方法会在每个线程第一次访问该变量时执行，并返回初始化的值。
     * 在这段代码中，`Lambda` 表达式中的初始化方法会执行以下操作：
     * - 创建一个 `Kryo` 对象 `kryo`
     * - 使用 `kryo.register()` 方法注册 `RpcResponse.class` 和 `RpcRequest.class` 类型，以便在序列化和反序列化时可以正确处理这些类型
     * - 设置 `kryo` 对象的引用模式为 `true`，允许引用对象的序列化和反序列化
     * - 设置 `kryo` 对象的注册非必需模式为 `false`，允许序列化和反序列化未注册的类
     * - 返回初始化后的 `kryo` 对象
     * 因此，`kryoThreadLocal` 是一个 `ThreadLocal` 对象，用于存储每个线程独立的 `Kryo` 对象，保证了线程之间的隔离性和线程安全性
     */
    private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.register(RpcResponse.class);
        kryo.register(RpcRequest.class);
        kryo.setReferences(true);
        kryo.setRegistrationRequired(false);
        return kryo;
    });

    /**
     * 序列化对象
     * @param obj
     * @return 字节数组
     */
    @Override
    public byte[] serialize(Object obj) {
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
             Output output = new Output(byteArrayOutputStream)) {
            Kryo kryo = kryoThreadLocal.get();
            kryo.writeObject(output, obj);
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (Exception e) {
            logger.error("序列化时有错误发生:", e);
            throw new SerializeException("序列化时有错误发生");
        }
    }

    /*
        Kryo 的 Input 和 Output 接收一个 InputStream 和 OutputStream，Kryo 通常完成字节数组和对象的转换，
        所以常用的输入输出流实现为 ByteArrayInputStream/ByteArrayOutputStream
    */

    /**
     * 反序列化
     * @param bytes
     * @param clazz
     * @return 对象
     */
    @Override
    public Object deserialize(byte[] bytes, Class<?> clazz) {
        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
             Input input = new Input(byteArrayInputStream)) {
            Kryo kryo = kryoThreadLocal.get();
            Object o = kryo.readObject(input, clazz);
            kryoThreadLocal.remove();
            return o;
        } catch (Exception e) {
            logger.error("反序列化时有错误发生:", e);
            throw new SerializeException("反序列化时有错误发生");
        }
    }

    @Override
    public int getCode() {
        return SerializerCode.valueOf("KRYO").getCode();
    }
}
